# _*_ coding:utf-8 _*_
# @Time  : 2022.08.06
# @Author: zizlee
import time
import json
import zipfile
import datetime
import requests
import pathlib
import pandas as pd
from itertools import groupby
from urllib3 import disable_warnings
from zizlee_position import get_dominant_price_position, get_contract_price_position
disable_warnings()

USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) " \
             "Chrome/78.0.3904.108 Safari/537.36"

VARIETY_EN = {
    'IF': 'IF',
    'IC': 'IC',
    'IH': 'IH',
    '生猪': 'LH',
    '沪深300': 'IF',
    '中证500': 'IC',
    '上证50': 'IH',
    '铜(CU)': 'CU',
    '铜': 'CU',
    '铜(BC)': 'BC',
    '短纤': 'PF',
    '铝': 'AL',
    '铅': 'PB',
    '花生': 'PK',
    '锌': 'ZN',
    '锡': 'SN',
    '镍': 'NI',
    '铁矿石': 'I',
    '热轧卷板': 'HC',
    '热卷': 'HC',
    '螺纹钢': 'RB',
    '螺纹': 'RB',
    '螺纹钢仓库': 'RB',
    '螺纹钢厂库': 'RB',
    '线材': 'WR',
    '不锈钢': 'SS',
    '不锈钢仓库': 'SS',
    '硅铁': 'SF',
    '硅锰': 'SM',
    '锰硅': 'SM',
    '焦煤': 'JM',
    '焦炭': 'J',
    '动力煤': 'ZC',
    '郑煤': 'ZC',
    '黄金': 'AU',
    '白银': 'AG',
    '大豆': 'A',
    '豆一': 'A',
    '豆二': 'B',
    '胶合板': 'BB',
    '豆粕': 'M',
    '豆油': 'Y',
    '棕榈油': 'P',
    '粳米': 'RR',
    '白糖': 'SR',
    '棉花': 'CF',
    '棉纱': 'CY',
    '苹果': 'AP',
    '红枣': 'CJ',
    '聚丙烯': 'PP',
    '聚氯乙烯': 'V',
    '聚乙烯': 'L',
    '鸡蛋': 'JD',
    '菜粕': 'RM',
    '菜籽粕': 'RM',
    '菜油': 'OI',
    '菜籽油': 'OI',
    '玉米': 'C',
    '淀粉': 'CS',
    'LLDPE': 'L',
    'PP': 'PP',
    'PVC': 'V',
    '苯乙烯': 'EB',
    '低硫燃料油仓库': 'LU',
    '低硫燃料油厂库': 'LU',
    '全乳胶': 'RU',
    '天然橡胶': 'RU',
    '橡胶': 'RU',
    '20号胶': 'NR',
    'STR20': 'NR',
    '甲醇': 'MA',
    '尿素': 'UR',
    '玻璃': 'FG',
    '纯碱': 'SA',
    '乙二醇': 'EG',
    'PTA': 'TA',
    '纸浆': 'SP',
    '纸浆仓库': 'SP',
    '纸浆厂库': 'SP',
    '沥青': 'BU',
    '沥青仓库': 'BU',
    '沥青厂库': 'BU',
    '石油沥青仓库': 'BU',
    '石油沥青厂库': 'BU',
    '纤维板': 'FB',
    '液化气': 'PG',
    'LPG': 'PG',
    '燃料油': 'FU',
    '液化石油气': 'PG',
    '原油': 'SC',
    '玉米淀粉': 'CS',
    '中质含硫原油': 'SC',
}


def split_number_en(ustring):
    return [''.join(list(g)) for k, g in groupby(ustring, key=lambda x: x.isdigit())]


def get_variety_en(name):
    variety_en = VARIETY_EN.get(name.replace('小计', '').replace(' ', ''), None)
    if variety_en is None:
        raise ValueError("大商所{}品种对应的代码不存在...".format(name))
    return variety_en


def full_width_to_half_width(ustring):
    """ 全角转半角 """
    reverse_str = ""
    for uchar in ustring:
        inside_code = ord(uchar)
        if inside_code == 12288:  # 全角空格直接转换
            inside_code = 32
        elif 65281 <= inside_code <= 65374:  # 全角字符（除空格）根据关系转化
            inside_code -= 65248
        else:
            pass
        reverse_str += chr(inside_code)
    return reverse_str


class DailyQuoteSpider(object):
    def __init__(self, date: datetime.datetime):
        self.quote_date = datetime.datetime.strptime(date.strftime('%Y-%m-%d'), '%Y-%m-%d')
        self.quote_date_string = self.quote_date.strftime('%Y%m%d')
        self.headers = {'User-Agent': USER_AGENT,
                        'Host': 'www.dce.com.cn'}
        self.quote_url = "http://www.dce.com.cn/publicweb/quotesdata/exportDayQuotesChData.html"
        self.quote_file = pathlib.Path("dce/daily/{}.xls".format(self.quote_date_string))
        self.rank_url = "http://www.dce.com.cn/publicweb/quotesdata/exportMemberDealPosiQuotesBatchData.html"
        self.rank_file = pathlib.Path("dce/rank/{}.zip".format(self.quote_date_string))
        self.receipt_url = "http://www.dce.com.cn/publicweb/quotesdata/wbillWeeklyQuotes.html"
        self.receipt_file = pathlib.Path("dce/receipt/{}.html".format(self.quote_date_string))

    # 获取日行情文件
    def get_quote_file(self):
        form_params = {
            "dayQuotes.variety": "all",
            "dayQuotes.trade_type": "0",
            "year": str(self.quote_date.year),
            "month": str(self.quote_date.month - 1),
            "day": self.quote_date.strftime("%d"),
            "exportFlag": "excel"
        }
        r = requests.post(self.quote_url, headers=self.headers, data=form_params)
        with open(self.quote_file, 'wb') as fp:
            fp.write(r.content)
        print('获取{}.DCE日行情数据完成！'.format(self.quote_date_string))

    # 获取日排名文件
    def get_rank_file(self):
        form_params = {
            'memberDealPosiQuotes.variety': 'a',
            'memberDealPosiQuotes.trade_type': '0',
            'year': str(self.quote_date.year),
            'month': str(self.quote_date.month - 1),
            'day': self.quote_date.strftime("%d"),
            'contract.contract_id': 'a2009',
            'contract.variety_id': 'a',
            'batchExportFlag': 'batch'
        }
        r = requests.post(self.rank_url, headers=self.headers, data=form_params)
        with open(self.rank_file, 'wb') as fp:
            fp.write(r.content)
        print('获取{}.DCE日排名数据完成！'.format(self.quote_date_string))

    # 获取日仓单文件
    def get_receipt_file(self):
        form_params = {
            'wbillWeeklyQuotes.variety': 'all',
            'year': str(self.quote_date.year),
            'month': str(self.quote_date.month - 1),
            'day': self.quote_date.strftime("%d"),
        }
        r = requests.post(self.receipt_url, headers=self.headers, data=form_params)
        with open(self.receipt_file, 'wb') as fp:
            fp.write(r.content)
        print('获取{}.DCE日仓单数据完成！'.format(self.quote_date_string))


class DailyQuoteParser(object):
    # SERVER_API = "http://127.0.0.1:8000/api/"
    SERVER_API = "https://210.13.218.130:9000/api/"

    def __init__(self, date: datetime.datetime):
        self.quote_date = datetime.datetime.strptime(date.strftime('%Y-%m-%d'), '%Y-%m-%d')
        self.quote_date_string = self.quote_date.strftime('%Y%m%d')

        self.quote_file = pathlib.Path("dce/daily/{}.xls".format(self.quote_date_string))
        self.rank_file = pathlib.Path("dce/rank/{}.zip".format(self.quote_date_string))
        self.receipt_file = pathlib.Path("dce/receipt/{}.html".format(self.quote_date_string))

        self.quote_save_url = self.SERVER_API + "exchange/dce/daily/?date={}".format(self.quote_date.strftime("%Y-%m-%d"))
        self.rank_save_url = self.SERVER_API + "exchange/dce/rank/?date={}".format(self.quote_date.strftime("%Y-%m-%d"))
        self.receipt_save_url = self.SERVER_API + "exchange/dce/receipt/"

        self.resolution_quote_file = pathlib.Path("resolution/{}/DCE_Quote.json".format(self.quote_date_string))
        self.resolution_rank_file = pathlib.Path("resolution/{}/DCE_Rank.json".format(self.quote_date_string))
        self.resolution_receipt_file = pathlib.Path("resolution/{}/DCE_Receipt.json".format(self.quote_date_string))
        self.resolution_receipt_file2 = pathlib.Path("resolution/{}/DCE_Receipt2.json".format(self.quote_date_string))
        resolution_folder = pathlib.Path("resolution/{}/".format(self.quote_date_string))
        if not resolution_folder.exists():
            resolution_folder.mkdir(parents=True)

    # 解析日行情数据
    def parse_quote_file(self):
        if not self.quote_file.exists():
            print("没有发现{}.DCE的行情文件,请先抓取数据!".format(self.quote_date_string))
            return
        if self.quote_date_string >= '20231009':
            xls_df = pd.read_excel(self.quote_file, thousands=',', skiprows=[0])
        else:
            xls_df = pd.read_excel(self.quote_file, thousands=',')
        # 判断表头格式
        headers = ['商品名称', '交割月份', '开盘价', '最高价', '最低价', '收盘价',
                   '前结算价', '结算价', '涨跌', '涨跌1', '成交量', '持仓量', '持仓量变化', '成交额']
        if self.quote_date.strftime('%Y-%m-%d') >= '2022-12-19':  # 1219名称换了
            headers[1] = '合约名称'
        if xls_df.columns.values.tolist() != headers:
            print("{}文件格式有误".format(self.quote_date_string))
            return

        # 选取无合计,小计,总计的行
        xls_df = xls_df[~xls_df["商品名称"].str.contains("小计|总计|合计")]

        if self.quote_date.strftime('%Y-%m-%d') >= '2022-12-19':  # 1219名称换了
            xls_df['交割月份'] = xls_df['合约名称'].apply(lambda x: split_number_en(x)[1])

        # 交个月份转为int再转为str
        xls_df["交割月份"] = xls_df["交割月份"].apply(lambda x: str(int(x)))
        # 修改商品名称为英文
        xls_df["商品名称"] = xls_df["商品名称"].apply(get_variety_en)
        # 加入日期
        int_date = int(self.quote_date.timestamp())
        xls_df["日期"] = [int_date for _ in range(xls_df.shape[0])]
        # 重置列头并重命名
        xls_df = xls_df.reindex(columns=["日期", "商品名称", "交割月份", "前结算价", "开盘价", "最高价", "最低价", "收盘价",
                                         "结算价", "涨跌", "涨跌1", "成交量", "持仓量", "持仓量变化", "成交额"])
        xls_df.columns = ["date", "variety_en", "contract", "pre_settlement", "open_price", "highest", "lowest",
                          "close_price", "settlement", "zd_1", "zd_2", "trade_volume", "empty_volume",
                          "increase_volume", "trade_price"]
        # 合约改为品种+交割月的形式
        xls_df["contract"] = xls_df["variety_en"] + xls_df["contract"]
        quote_data = xls_df.to_dict(orient='records')
        print('----- 解析{}.DCE行情文件成功,数量:{} -----'.format(self.quote_date_string, len(quote_data)))
        if len(quote_data) > 1:
            # 保存到本地文件
            with open(self.resolution_quote_file, 'w', encoding='utf8') as fp:
                json.dump(quote_data, fp, indent=2, ensure_ascii=False)

    # 解析日排名数据
    def parse_rank_file(self):
        if not self.rank_file.exists():
            print("没有发现{}.DCE的排名文件,请先抓取数据!".format(self.quote_date_string))
            return
        # 解压文件的缓存目录
        cache_folder = pathlib.Path('dce/rank/cache/{}/'.format(self.quote_date_string))
        # 解压文件到文件夹
        zip_file = zipfile.ZipFile(self.rank_file)
        zip_list = zip_file.namelist()
        for filename in zip_list:
            # filename = filename.encode('cp437').decode('gbk')  # 这样做会无法提取文件。遂修改源代码
            zip_file.extract(filename, cache_folder)  # 循环解压文件到指定目录
        zip_file.close()
        # 取解压后的文件夹下的文件，逐个读取内容解析得到最终的数据集
        value_df = self._parser_variety_rank(cache_folder)
        if not value_df.empty:
            # 填充空值(合并后产生空值)
            value_df = value_df.fillna('-')
            # 将数据需要为int列转为int
            value_df["rank"] = value_df["rank"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["trade"] = value_df["trade"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["trade_increase"] = value_df["trade_increase"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["long_position"] = value_df["long_position"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["long_position_increase"] = value_df["long_position_increase"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["short_position"] = value_df["short_position"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            value_df["short_position_increase"] = value_df["short_position_increase"].apply(lambda x: 0 if x == '-' else int(x.replace(',', '')))
            # 日期转为整形时间戳
            value_df['date'] = value_df['date'].apply(lambda x: int(datetime.datetime.strptime(x, '%Y%m%d').timestamp()))

        rank_data = value_df.to_dict(orient='records')
        print('----- 解析{}.DCE排名文件成功,数量:{} -----'.format(self.quote_date_string, len(rank_data)))
        if len(rank_data) > 1:
            # 保存到本地文件
            with open(self.resolution_rank_file, 'w', encoding='utf8') as fp:
                json.dump(rank_data, fp, indent=2, ensure_ascii=False)

    # 解析日仓单文件
    def parse_receipt_file(self):
        if not self.receipt_file.exists():
            print("没有发现{}.DCE的仓单文件,请先抓取数据!".format(self.quote_date_string))
            return
        html_df = pd.read_html(self.receipt_file, encoding='utf-8')[0]
        if html_df.columns.values.tolist() != ['品种', '仓库/分库', '可选提货地点', '昨日仓单量', '今日仓单量', '增减']:
            print("{}.DCE的数据格式出现错误,解析失败!".format(self.quote_date_string))
            return
        # 修改列头
        html_df.columns = ["VARIETY", "WARENAME", "ADDR", "YRECEIPT", "TRECEIPT", "INCREATE"]
        # 填充nan为上一行数据
        html_df.fillna(method='ffill', inplace=True)
        # # 去除品种含小计,总计等行
        # html_df = html_df[~html_df['VARIETY'].str.contains('总计|小计|合计')]
        # 选取含有小计的行(20210226改：取品种的结果只能取这个才准确)
        html_df = html_df[html_df['VARIETY'].str.contains('小计')]
        # 仓库简称处理
        html_df["WARENAME"] = html_df["WARENAME"].apply(full_width_to_half_width)
        # 增加品种代码
        html_df["VARIETYEN"] = html_df["VARIETY"].apply(get_variety_en)
        # 增加今日时间
        html_df["DATE"] = self.quote_date_string
        # 重置索引
        result_df = html_df.reindex(columns=["WARENAME", "VARIETYEN", "DATE", "TRECEIPT", "INCREATE"])
        # 品种分组求和
        sum_df = result_df.groupby(by=['VARIETYEN'], as_index=False)[['TRECEIPT', 'INCREATE']].sum()
        sum_df["DATE"] = int(self.quote_date.timestamp())
        sum_df.columns = ["variety_en", "receipt", "increase", "date"]
        sum_df = sum_df.reindex(columns=["date", "variety_en", "receipt", "increase"])
        receipt_data = sum_df.to_dict(orient='records')
        print('----- 解析{}.DCE仓单文件成功,数量:{} -----'.format(self.quote_date_string, len(receipt_data)))
        if len(receipt_data) > 1:
            # 保存到本地文件
            with open(self.resolution_receipt_file, 'w', encoding='utf8') as fp:
                json.dump(receipt_data, fp, indent=2, ensure_ascii=False)

    # 解析日仓单文件2(带仓库名的数据)
    def parse_receipt_file2(self):
        if not self.receipt_file.exists():
            print("没有发现{}.DCE的仓单文件,请先抓取数据!".format(self.quote_date_string))
            return
        # 解析文件
        html_df = pd.read_html(self.receipt_file, encoding='utf-8')[0]
        if html_df.columns.values.tolist() != ['品种', '仓库/分库', '可选提货地点', '昨日仓单量', '今日仓单量', '增减']:
            print("DCE的仓单数据格式出现错误,解析失败!")
            return
        # 修改列头
        html_df.columns = ["VARIETY", "WARENAME", "ADDR", "YRECEIPT", "TRECEIPT", "INCREATE"]

        # 增加一行标记，VARIETY=nan时，不计入总计的标记, ex_total=0是计入小计的
        html_df['EX_TOTAL'] = html_df['VARIETY'].apply(lambda x: 1 if pd.isnull(x) else 0)

        # 填充nan为上一行数据
        html_df.fillna(method='ffill', inplace=True)
        # 去除品种含小计,总计等行
        html_df = html_df[~html_df['VARIETY'].str.contains('总计|小计|合计')]
        # # 选取含有小计的行(20210226改：取品种的结果只能取这个才准确)
        # html_df = html_df[html_df['VARIETY'].str.contains('小计')]
        # 仓库简称处理
        html_df["WARENAME"] = html_df["WARENAME"].apply(full_width_to_half_width)
        # 增加品种代码
        html_df["VARIETYEN"] = html_df["VARIETY"].apply(get_variety_en)
        # 重置索引
        result_df = html_df.reindex(columns=["WARENAME", "VARIETYEN", "TRECEIPT", "INCREATE", 'EX_TOTAL'])
        result_df.columns = ["warehouse", "variety_en", "receipt", "increase", 'ex_total']
        # 增加日期字段df['receipt_date'] = [data_str for _ in range(df.shape[0])]
        result_df['receipt_date'] = self.quote_date.strftime('%Y-%m-%d')
        receipt_data = result_df.to_dict(orient='records')
        print('----- 解析{}.DCE仓单文件2成功,数量:{} -----'.format(self.quote_date_string, len(receipt_data)))
        if len(receipt_data) > 1:
            # 保存到本地文件
            with open(self.resolution_receipt_file2, 'w', encoding='utf8') as fp:
                json.dump(receipt_data, fp, indent=2, ensure_ascii=False)

    # 解析解压出来的文件
    def _parser_variety_rank(self, cache_folder):
        all_data_df = pd.DataFrame(columns=["date", "variety_en", "contract", "rank",
                                            "trade_company", "trade", "trade_increase",
                                            "long_position_company", "long_position", "long_position_increase",
                                            "short_position_company", "short_position", "short_position_increase"])

        filename_list = pathlib.Path(cache_folder).glob('*.txt')
        for contract_filename in filename_list:
            contract_file_path = contract_filename
            contract_filename = str(contract_filename)
            # contract_file_path = cache_folder.joinpath(contract_filename)
            message_list = contract_filename.split('_')
            c_date = pathlib.Path(message_list[0]).name  # 得到日期
            contract = message_list[1].upper()  # 得到合约
            variety_en = split_number_en(message_list[1])[0].upper()  # 得到合约代码
            if self.quote_date.strftime("%Y%m%d") < '20160101':  # 解析20160101之前的数据,使用gbk解码
                contract_df = pd.read_table(contract_file_path, encoding="gbk")
            else:  # 解析20160101之后的数据,使用utf8解码
                contract_df = pd.read_table(contract_file_path, sep=' ')
            # 初始化3个容器和对应的置入信号
            # 1 成交量容器
            trade_values = []
            trade_append = False
            # 2 买单量容器
            long_position_values = []
            long_position_append = False
            # 卖单量容器
            short_position_values = []
            short_position_append = False
            for df_row in contract_df.itertuples():
                row_list = df_row[1].split()
                if row_list[0] in ["总计", "会员类别", "期货公司会员",
                                   "非期货公司会员"]:  # 去除总计行否则后面添加报错， 20131126以后多了"会员类别", "期货公司会员", "非期货公司会员"数据
                    continue
                # 相应的数据放入对应的容器中
                if row_list == ['名次', '会员简称', '成交量', '增减']:
                    # 打开成交量容器
                    trade_append = True
                    long_position_append = False
                    short_position_append = False
                    continue
                if row_list == ['名次', '会员简称', '持买单量', '增减']:
                    # 打开买单量容器
                    trade_append = False
                    long_position_append = True
                    short_position_append = False
                    continue
                if row_list == ['名次', '会员简称', '持卖单量', '增减']:
                    # 打开卖单量容器
                    trade_append = False
                    long_position_append = False
                    short_position_append = True
                    continue
                if trade_append and 1 <= int(row_list[0]) <= 20:
                    trade_values.append(
                        {
                            "date": c_date,
                            "contract": contract,
                            "variety_en": variety_en,
                            "rank": row_list[0],
                            "trade_company": row_list[1],
                            "trade": row_list[2],
                            "trade_increase": row_list[3]
                        }
                    )
                if long_position_append and 1 <= int(row_list[0]) <= 20:
                    long_position_values.append(
                        {
                            "date": c_date,
                            "contract": contract,
                            "variety_en": variety_en,
                            "rank": row_list[0],
                            "long_position_company": row_list[1],
                            "long_position": row_list[2],
                            "long_position_increase": row_list[3]

                        }
                    )
                if short_position_append and 1 <= int(row_list[0]) <= 20:
                    short_position_values.append(
                        {
                            "date": c_date,
                            "contract": contract,
                            "variety_en": variety_en,
                            "rank": row_list[0],
                            "short_position_company": row_list[1],
                            "short_position": row_list[2],
                            "short_position_increase": row_list[3]
                        }
                    )
            # 得到3个数据集
            # print("成交量数据集")
            # for t in trade_values:
            #     print(t)
            # print("买单量数据集")
            # for l in long_position_values:
            #     print(l)
            # print("卖单量数据集")
            # for s in short_position_values:
            #     print(s)
            # 将数据集转为DataFrame
            columns_list = ["date", "contract", "variety_en", "rank"]
            trade_df = pd.DataFrame(trade_values, columns=columns_list + ["trade_company", "trade", "trade_increase"])
            long_position_df = pd.DataFrame(long_position_values,
                                            columns=columns_list + ["long_position_company", "long_position",
                                                                    "long_position_increase"])
            short_position_df = pd.DataFrame(short_position_values,
                                             columns=columns_list + ["short_position_company", "short_position",
                                                                     "short_position_increase"])
            # 横向合并
            contract_result_df = pd.merge(trade_df, long_position_df, on=["date", "contract", "variety_en", "rank"],
                                          how="outer")
            contract_result_df = pd.merge(contract_result_df, short_position_df,
                                          on=["date", "contract", "variety_en", "rank"], how="outer")
            # 将数据纵向合并到总DataFrame
            all_data_df = pd.concat([all_data_df, contract_result_df])
        return all_data_df

    # 读取解析好的行情数据
    def read_daily_quote(self):
        if not self.resolution_quote_file.exists():
            raise ValueError('{}.DCE行情数据文件不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_quote_file, 'r', encoding='utf8') as fp:
            quote_data = json.load(fp)
        return quote_data

    # 读取解析好的排名数据
    def read_daily_rank(self):
        if not self.resolution_rank_file.exists():
            raise ValueError('{}.DCE排名数据文件不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_rank_file, 'r', encoding='utf8') as fp:
            rank_data = json.load(fp)
        return rank_data

    # 读取解析好的仓单数据
    def read_daily_receipt(self):
        if not self.resolution_rank_file.exists():
            raise ValueError('{}.DCE仓单数据文件不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_receipt_file, 'r', encoding='utf8') as fp:
            receipt_data = json.load(fp)
        return receipt_data

    # 读取解析好的仓单数据2
    def read_daily_receipt2(self):
        if not self.resolution_receipt_file2.exists():
            raise ValueError('{}.DCE仓单数据文件2不存在,请先解析保存！'.format(self.quote_date_string))
        # 读取本地文件
        with open(self.resolution_receipt_file2, 'r', encoding='utf8') as fp:
            receipt_data = json.load(fp)
        return receipt_data

    # 保存日行情数据
    def save_daily_quote(self):
        quote_data = self.read_daily_quote()
        try:
            r = requests.post(self.quote_save_url, json=quote_data, verify=False)
            print(r.json())
        except Exception as e:
            print('保存{}.DCE行情数据失败了:{}'.format(self.quote_date_string, e))
        time.sleep(1)

    # 保存日排名数据
    def save_daily_rank(self):
        rank_data = self.read_daily_rank()
        try:
            r = requests.post(self.rank_save_url, json=rank_data, verify=False)
            print(r.json())
        except Exception as e:
            print('保存{}.DCE排名数据失败了:{}'.format(self.quote_date_string, e))
        time.sleep(1)

    # 保存仓单数据
    def save_daily_receipt(self):
        receipt_data = self.read_daily_receipt()
        try:
            r = requests.post(self.receipt_save_url, json=receipt_data, verify=False)
            print(r.json())
        except Exception as e:
            print('保存{}.DCE仓单数据失败了:{}'.format(self.quote_date_string, e))
        time.sleep(1)

    # 总表接口:保存行情数据，所有交易所数据放一个表的
    def new_save_quote(self):
        quote_data = self.read_daily_quote()
        df = pd.DataFrame(quote_data)
        df['quotes_date'] = df['date'].apply(lambda x: datetime.datetime.fromtimestamp(x).strftime('%Y-%m-%d'))
        df.rename(columns={'empty_volume': 'position_volume'}, inplace=True)

        save_url = self.SERVER_API + 'dat/quotes/daily-quotes/'
        try:
            r = requests.post(save_url, json=df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.DCE行情数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.DCE行情数据成功:{},message:{}'.format(self.quote_date_string, r_data['count'], r_data['message']))
        time.sleep(1)

    # 总表接口:保存排名数据，所有交易所数据放一个表的
    def new_save_rank(self):
        rank_data = self.read_daily_rank()
        df = pd.DataFrame(rank_data)
        df['rank_date'] = df['date'].apply(lambda x: datetime.datetime.fromtimestamp(x).strftime('%Y-%m-%d'))
        save_url = self.SERVER_API + 'dat/rank/daily-rank/'
        try:
            r = requests.post(save_url, json=df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.DCE排名数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.DCE排名数据成功:{},message:{}'.format(self.quote_date_string, r_data['count'], r_data['message']))
        time.sleep(1)

    # 总表接口:保存仓单数据，所有交易所数据放一个表的
    def new_save_receipt(self):
        receipt_data = self.read_daily_receipt2()
        df = pd.DataFrame(receipt_data)
        save_url = self.SERVER_API + 'dat/receipt/daily-receipt/'
        try:
            r = requests.post(save_url, json=df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.DCE仓单数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.DCE仓单数据成功:{},message:{}'.format(self.quote_date_string, r_data['count'], r_data['message']))
        time.sleep(1)

    def run_positions_handler(self):
        # 日行情
        quote_data = self.read_daily_quote()
        quotes_df = pd.DataFrame(quote_data)
        quotes_df.rename(columns={'date': 'quotes_ts', 'empty_volume': 'position_volume'}, inplace=True)
        # 日排名
        rank_data = self.read_daily_rank()
        rank_df = pd.DataFrame(rank_data)
        rank_df.rename(columns={'date': 'rank_ts'}, inplace=True)
        if quotes_df.empty or rank_df.empty:
            print('{}.DCE行情或持仓数据为空!'.format(self.quote_date_string))
            return
        quotes_df = quotes_df[
            ['quotes_ts', 'variety_en', 'contract', 'close_price', 'trade_volume', 'position_volume']]
        rank_df = rank_df[['rank_ts', 'variety_en', 'contract', 'long_position', 'short_position']]
        dominant_df = get_dominant_price_position(quotes_df.copy(), rank_df.copy())
        # 获取合约的持仓数据
        contract_df = get_contract_price_position(quotes_df.copy(), rank_df.copy())
        final_df = pd.concat([dominant_df, contract_df])
        for col in ['close_price', 'position_price', 'position_volume', 'long_position', 'short_position']:
            final_df[col] = final_df[col].apply(lambda x: int(x) if int(x) == float(x) else round(x, 4))
        final_df.sort_values(by='contract', inplace=True)
        # for row in final_df.to_dict(orient='records'):
        #     print(row)
        # 保存到服务器
        save_url = self.SERVER_API + 'dsas/price-position/'
        try:
            r = requests.post(save_url, json=final_df.to_dict(orient='records'), verify=False)
            r_data = r.json()
        except Exception as e:
            print('新版保存{}.DCE持仓价格数据失败了:{}'.format(self.quote_date_string, e))
        else:
            print('新版保存{}.DCE持仓价格数据成功:{}'.format(self.quote_date_string, r_data['message']))


if __name__ == '__main__':
    SPIDER = 1
    PARSER = 0
    delta_days = -61
    if SPIDER:
        spider = DailyQuoteSpider(date=datetime.datetime.today() + datetime.timedelta(days=delta_days))
        spider.get_quote_file()    # 获取日行情数据保存为文件
        spider.get_rank_file()     # 获取日排名数据保存为文件
        spider.get_receipt_file()  # 获取日仓单数据保存为文件
    else:
        parser = DailyQuoteParser(date=datetime.datetime.today() + datetime.timedelta(days=delta_days))
        if PARSER:
            parser.parse_quote_file()
            parser.parse_rank_file()
            parser.parse_receipt_file()
            parser.parse_receipt_file2()
        else:
            # --- 分库分表 ---
            parser.save_daily_quote()    # 保存日行情
            parser.save_daily_rank()     # 保存日排名
            parser.save_daily_receipt()  # 保存日仓单
            # --- 总表 ---
            parser.new_save_quote()      # 保存解析的行情数据
            parser.new_save_rank()       # 保存解析的持仓数据
            parser.new_save_receipt()    # 保存仓单的数据(含仓库简称)
            # -- 净持仓 ---
            parser.run_positions_handler()  # 处理保存净持仓数据
